feat(clickhouse): add StagingStore abstraction for GCS staging support#4200
Conversation
|
|
❌ 2 Tests Failed:
View the top 2 failed test(s) by shortest run time
To view more test analytics, go to the Test Analytics Dashboard |
🔄 Flaky Test DetectedAnalysis: Two unrelated flaky failures: a 1-nanosecond time precision mismatch in Test_Types_CH and a transient PostgreSQL catalog connection termination (SQLSTATE 57P01) in TestCancelTableAdditionRemoveAddRemove, both caused by CI environment non-determinism rather than code regressions. ✅ Automatically retrying the workflow |
🔄 Flaky Test DetectedAnalysis: The single failing test ✅ Automatically retrying the workflow |
🔄 Flaky Test DetectedAnalysis: Both failing tests ( ✅ Automatically retrying the workflow |
🔄 Flaky Test DetectedAnalysis: The primary failure is a transient Temporal server connectivity timeout ("context deadline exceeded") causing test infrastructure to be unreachable, with a secondary BigQuery test timeout — both are classic flaky CI infrastructure failures unrelated to code changes. ✅ Automatically retrying the workflow |
🔄 Flaky Test DetectedAnalysis: The job failed during Docker container startup because port 49002 was already in use on the CI runner host (toxiproxy container networking error), a transient infrastructure issue unrelated to any code change. ✅ Automatically retrying the workflow |
❌ Test FailureAnalysis: Real build failure: |
whites11
force-pushed
the
feat/gcs-staging-store
branch
from
84f57d7 to
53ab53d
Compare
❌ Test FailureAnalysis: Deterministic compile error: |
whites11
force-pushed
the
feat/gcs-staging-store
branch
from
53ab53d to
d631eef
Compare
🔄 Flaky Test DetectedAnalysis: The e2e test ✅ Automatically retrying the workflow |
whites11
force-pushed
the
feat/gcs-staging-store
branch
3 times, most recently
from
52ad279 to
496c92f
Compare
❌ Test FailureAnalysis: Deterministic compile error: method peerDBOCFWriter.WriteRecordsToS3 is declared twice in connectors/utils/avro_writer.go (lines 96 and 170), introduced by the feat/gcs-staging-store branch. |
❌ Test FailureAnalysis: Build fails with a Go compilation error — |
🔄 Flaky Test DetectedAnalysis: The e2e test ✅ Automatically retrying the workflow |
…ging Introduce a StagingStore interface that abstracts cloud storage operations for ClickHouse CDC/snapshot staging. This enables GCS support via signed URLs and prepares for Azure Blob Storage support. Changes: - New StagingStore interface with Upload, TableFunctionExpr, DeletePrefix, Validate, and KeyPrefix methods - S3StagingStore: extracts existing S3 logic (zero behavior change) - GCSStagingStore: uses native GCS SDK with Workload Identity auth and ClickHouse url() table function with signed URLs - WriteRecordsToStaging: cloud-agnostic Avro upload method - PEERDB_CLICKHOUSE_STAGING_PROVIDER env var (s3 default, gcs) - PEERDB_CLICKHOUSE_GCS_BUCKET_NAME env var for GCS bucket config The S3 path is completely unchanged — GCS is additive only. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- Reorder S3StagingStore fields to fix fieldalignment (64→56 pointer bytes) - Use require.Empty instead of require.Equal(t, "", ...) per testifylint - Remove unused newValidationCheckKey function and its uuid import Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…nv var Add provider-agnostic PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME that works with PEERDB_CLICKHOUSE_STAGING_PROVIDER to configure staging storage. GCS uses the unified var exclusively. S3 falls back to the legacy PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME when the unified var is not set. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
The StagingStore-based upload method had the same name as the existing S3-specific method, causing a compile error. Rename to WriteRecordsToStaging to distinguish the generic staging interface from the legacy S3 path.
whites11
force-pushed
the
feat/gcs-staging-store
branch
from
496c92f to
553bbb7
Compare
Move createStagingStore, createS3StagingStore, and createGCSStagingStore into a dedicated file to keep the clickhouse.go diff minimal. Remove unused ValidateS3 and ValidateStaging functions.
…S3StagingStore Pass ClickHouse version to createStagingStore so the S3-specific session token validation lives alongside the S3 credential setup rather than in NewClickHouseConnector.
🔄 Flaky Test DetectedAnalysis: The e2e test suite in the ✅ Automatically retrying the workflow |
…Staging Remove the WriteRecordsToStaging wrapper from avro_writer.go and call StagingStore.Upload directly in the ClickHouse connector. This avoids adding a new method to avro_writer.go and keeps the upload logic colocated with the ClickHouse-specific staging setup.
🔄 Flaky Test DetectedAnalysis: The e2e test suite hit the 900-second timeout limit (ran for 900.748s), indicating a timeout-induced failure rather than a code regression — all other test packages passed cleanly. ✅ Automatically retrying the workflow |
🔄 Flaky Test DetectedAnalysis: The test failed due to ✅ Automatically retrying the workflow |
🔄 Flaky Test DetectedAnalysis: TestMongoClickhouseSuite/Test_Json_Disabled failed with ClickHouse error code 60 (UNKNOWN_TABLE) during the initial-load polling loop — the test queried the destination table before it was created, a classic e2e timing race condition. ✅ Automatically retrying the workflow |
Merge newS3StagingStoreFromConfig + newS3StagingStore into one newS3StagingStore. Same for GCS. Each staging_*.go file now has: struct → constructor → methods. staging.go is just the router.
🔄 Flaky Test DetectedAnalysis: The test TestApiPg/TestTableAdditionWithoutInitialLoad failed during teardown due to a race condition: the test logic completed successfully, but cleanup couldn't drop the replication slot because a background replication worker process (PID 16388) still held it active when teardown ran. ✅ Automatically retrying the workflow |
🔄 Flaky Test DetectedAnalysis: The test failed due to a PostgreSQL catalog connection being forcibly terminated mid-run (FATAL: terminating connection due to administrator command, SQLSTATE 57P01), which is a transient CI infrastructure issue unrelated to the code changes. ✅ Automatically retrying the workflow |
🔄 Flaky Test DetectedAnalysis: E2E tests failed due to Temporal client connection timeouts ("context deadline exceeded") and snapshot-stuck timing issues, with different tests failing on different runners and all running near the 900s timeout limit — classic signs of infrastructure flakiness rather than a code regression. ✅ Automatically retrying the workflow |
🔄 Flaky Test DetectedAnalysis: Test_Types_CH fails due to a 1-microsecond rounding difference in QValueTime comparison (e.g. 9h20m44.867835s vs 9h20m44.867834s) that is time-dependent — the actual timestamp values differ across runs, indicating this is a precision edge case triggered by whatever microsecond boundary ✅ Automatically retrying the workflow |
🔄 Flaky Test DetectedAnalysis: The test ✅ Automatically retrying the workflow |
🔄 Flaky Test DetectedAnalysis: TestApiMongo/TestCancelAddCancel failed due to a transient infrastructure issue — the CI catalog PostgreSQL connection was terminated by an administrator command (SQLSTATE 57P01) mid-test, not a code defect. ✅ Automatically retrying the workflow |
…ouseS3Credentials Per @jgao54 review on PR #4200: - avro_sync.go: include runtime/debug.Stack() in the panic-recovery error so the stacktrace is preserved (#4200 (comment)) - utils/aws.go: delete now-unused ClickHouseS3Credentials struct (#4200 (comment))
Code Review1 issue found. Checked for bugs and CLAUDE.md compliance. Bug: GCS storage.Client is never closed — resource leak File: flow/connectors/clickhouse/staging_store.go (interface definition) / flow/connectors/clickhouse/staging_gcs.go (line 51) The gcsStagingStore creates a *storage.Client in newGCSStagingStore, but this client is never closed. The StagingStore interface has no Close() method, and ClickHouseConnector.Close() only closes the ClickHouse database connection. The GCS storage.Client holds HTTP connections and background goroutines that should be released (its godoc explicitly states the caller must call Close). Each ClickHouseConnector created with staging_provider=gcs will leak a client. The S3 path does not have this issue since it creates ephemeral S3 clients on demand. Suggested fix: Add a Close() error method to the StagingStore interface, implement it on gcsStagingStore to call g.client.Close(), add a no-op on s3StagingStore, and call c.staging.Close() from ClickHouseConnector.Close(). |
🔄 Flaky Test DetectedAnalysis: The e2e test suite hit the 15-minute global timeout ( ✅ Automatically retrying the workflow |
🔄 Flaky Test DetectedAnalysis: The e2e test ✅ Automatically retrying the workflow |
🔄 Flaky Test DetectedAnalysis: The e2e test ✅ Automatically retrying the workflow |
## Summary Restores the ability to retain ClickHouse staging avro files after a snapshot/QRep flow completes, addressing a silent behavior change introduced in #4200. ## Background — the regression #4200 (`StagingStore` abstraction) refactored the ClickHouse connector so avro **upload** and staging **cleanup** both key off the staging store's real bucket prefix (`staging.KeyPrefix()`). The PR was described as "zero behavior change" for the S3 path, but the cleanup path actually changed behavior: - **Before:** `CleanupQRepFlow` keyed off `config.StagingPath` (= `SnapshotStagingPath`), guarded by `strings.HasPrefix(stagingPath, "s3://")`. For ClickHouse mirrors, `SnapshotStagingPath` is typically empty (the bucket comes from `PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME`), so the guard was false and **cleanup was a no-op** — staging avro files were retained in the bucket. - **After:** cleanup keys off the same real prefix that uploads use, so it now always deletes `<staging-prefix>/<flowJobName>` once the flow completes. Deployments that upgraded from 0.36.18 → 0.36.24 and had downstream pipelines consuming those staging avro artifacts saw the files disappear as soon as the mirror's snapshot completed. ## Change Cleanup remains the default — it is the correct behavior and avoids unbounded staging growth. This PR adds an opt-out: - New dynconf setting `PEERDB_CLICKHOUSE_SKIP_STAGING_CLEANUP` (bool, default `false`). - When enabled, `CleanupQRepFlow` logs and returns early without deleting staging objects. ## Notes - The staging bucket is PeerDB-internal scratch space; relying on retained files is fragile. The recommended long-term path for consuming avro artifacts is a dedicated PG→S3 QRep mirror. This flag is an escape hatch for affected deployments. - Separately, there is a latent mismatch when `PEERDB_S3_UUID_PREFIX` is enabled: upload keys become `<prefix>/<uuid>/<flowJobName>/...` while cleanup deletes `<prefix>/<flowJobName>`, so cleanup misses those objects. Out of scope here but worth tracking. ## Test plan - [x] `go build ./connectors/clickhouse/... ./internal/...` - [x] `go vet ./connectors/clickhouse/` - [ ] Manual: run a CH snapshot with `PEERDB_CLICKHOUSE_SKIP_STAGING_CLEANUP=true` and confirm staging avro files remain after completion; with default, confirm they are deleted. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
4 participants
Summary
Resolves https://linear.app/clickhouse/issue/CLP-629/add-gcs-support-for-staging-bucket
Adds GCS support for ClickHouse CDC/snapshot staging, alongside the existing S3 path.
What changed
StagingStoreinterface (utils/staging.go) — abstracts Upload, TableFunctionExpr, DeletePrefix, Validate across cloud providersS3StagingStore(utils/staging_s3.go) — extracts existing S3 logic, zero behavior changeGCSStagingStore(utils/staging_gcs.go) — native GCS SDK with Workload Identity auth; ClickHouse reads viaurl()table function with signed URLsstaging.go(clickhouse connector) —createStagingStoreroutes to S3 or GCS based onPEERDB_CLICKHOUSE_STAGING_PROVIDERenv varclickhouse.go— minimal diff: addsstagingfield, replaces inline S3 setup withcreateStagingStore()call, wraps session-token check in nil guard for non-S3 providersavro_sync.go— callsstore.Upload()directly instead of legacyWriteRecordsToS3dynamicconf.go—PeerDBClickHouseStagingBucketNamefalls back to legacyPEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME; addsPEERDB_CLICKHOUSE_STAGING_PROVIDERconfigHow GCS staging works
url('https://storage.googleapis.com/...?X-Goog-Signature=...', 'Avro')Env vars
PEERDB_CLICKHOUSE_STAGING_PROVIDERs3s3orgcsPEERDB_CLICKHOUSE_STAGING_BUCKET_NAMEPEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAMEfor S3Test plan
S3StagingStore(construction, table function expr, endpoint override, session tokens)🤖 Generated with Claude Code